package com.study.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;

/**
 * @author Mayjean
 * @Date2020/9/2 21:41
 */
public class SimpleCanalClientExample {

    public static void main(String args[]) {
        // 创建链接  AddressUtils.getHostIp()
        SocketAddress socketAddress = new InetSocketAddress("192.168.10.162", 11111);
        String username = "root";
        String password = "123456";
        CanalConnector connector = CanalConnectors.newSingleConnector(socketAddress, "example", username, password);
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            final int totalEmptyCount = 600;
            while (emptyCount < totalEmptyCount) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    //System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    //System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message);
                }

                // 提交确认
                connector.ack(batchId);
                // 处理失败, 回滚数据
                // connector.rollback(batchId);
            }

            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(Message message) {
        for (CanalEntry.Entry entry : message.getEntries()) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error,data:" + entry.toString(), e);
            }

            CanalEntry.EventType eventType = rowChange.getEventType();
            CanalEntry.Header header = entry.getHeader();
            System.out.println(String.format("================binlog[%s:%s],name[%s,%s],eventType : %s sql:%s",
                    header.getLogfileName(), header.getLogfileOffset(), header.getSchemaName(), header.getTableName(),
                    eventType, rowChange.getSql()));
            String databaseName = header.getSchemaName();
            String tableName = header.getTableName();


            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                printColumn(eventType,databaseName,tableName,rowData,rowChange);
//                if (eventType == CanalEntry.EventType.DELETE) {
//                    printColumn(eventType,databaseName,tableName,rowData.getBeforeColumnsList(),rowChange);
//                } else if (eventType == CanalEntry.EventType.INSERT) {
//                    printColumn(eventType,databaseName,tableName,rowData.getAfterColumnsList(),rowChange);
//                } else {
//                    System.out.println("-------before");
//                    printColumn(eventType,databaseName,tableName,rowData.getBeforeColumnsList(),rowChange);
//                    System.out.println("-------after");
//                    printColumn(eventType,databaseName,tableName,rowData.getAfterColumnsList(),rowChange);
//                }
            }

            continue;
        }
    }

    private static void printColumn(CanalEntry.EventType eventType,String databaseName,String tableName
            , CanalEntry.RowData rowData,CanalEntry.RowChange rowChange) {
        StringBuilder sql = new StringBuilder(128);
        if(eventType== CanalEntry.EventType.INSERT) {
            sql.append("insert into "+databaseName+"."+tableName);
            sql.append(" set ");
            List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
            for (CanalEntry.Column column : columns) {
                if("id".equals(column.getName())){
                    //TODO 自增列
                }
                if(column.getUpdated()) {
                    sql.append(column.getName() + "='" + column.getValue() + "'");
                }
            }
        }
        else if(eventType == CanalEntry.EventType.UPDATE) {
            sql.append("update "+databaseName+"."+tableName);
            sql.append(" set ");
            List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
            String whereStr = "";
            for (CanalEntry.Column column : columns) {
                if("id".equals(column.getName())){
                    whereStr = " where id="+column.getValue();
                    continue;
                }

                sql.append(column.getName() + "='" + column.getValue() + "',");
            }
            sql.deleteCharAt(sql.length()-1);
            sql.append(whereStr);
        }
        else if(eventType== CanalEntry.EventType.DELETE) {
            sql.append("delete from "+databaseName+"."+tableName);
            sql.append(" where ");
            List<CanalEntry.Column> columns = rowData.getBeforeColumnsList();
            for (CanalEntry.Column column : columns) {
                if("id".equals(column.getName())){
                    // to do nothing
                }
                sql.append(column.getName() + "='" + column.getValue() + "',");
            }
            sql.deleteCharAt(sql.length()-1);
        }
        else if(eventType== CanalEntry.EventType.CREATE ||eventType== CanalEntry.EventType.ALTER
            || eventType== CanalEntry.EventType.TRUNCATE) {
            sql.append(rowChange.getSql());
        }
        else if(eventType== CanalEntry.EventType.ERASE) {

        }
        else if(eventType== CanalEntry.EventType.QUERY) {

        }
        else if(eventType== CanalEntry.EventType.RENAME) {
            sql.append(rowChange.getSql());
        }
        else if(eventType== CanalEntry.EventType.CINDEX || eventType== CanalEntry.EventType.DINDEX) {
            sql.append(rowChange.getSql());
        }
        else if(eventType== CanalEntry.EventType.GTID) {

        }
        else if(eventType== CanalEntry.EventType.XACOMMIT) {

        }
        else if(eventType== CanalEntry.EventType.XAROLLBACK) {

        }
        else if(eventType== CanalEntry.EventType.MHEARTBEAT) {

        }
        else{
            System.out.println("未知的类型"+eventType);
            return;
        }

        System.out.println("SQL:"+sql.toString());
//        for (CanalEntry.Column column : columns) {
//            System.out.println(column.getName() + ":" + column.getValue() + "    update=" + column.getUpdated());
//        }
    }
}